package com.bigdata.spark.sql

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.expressions.Aggregator
import org.apache.spark.sql._

import java.util.Properties

object Spark04_SparkSQL_JDBC {

    def main(args: Array[String]): Unit = {

        // TODO 创建SparkSQL的运行环境
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkSQL")
        val spark = SparkSession.builder().config(sparkConf).getOrCreate()
        import spark.implicits._

        // 读取MySQL数据
        val df = spark.read
            .format("jdbc")
            .option("url", "jdbc:mysql://hadoop102:3306/atcper")
            .option("driver", "com.mysql.cj.jdbc.Driver")
            .option("user", "root")
            .option("password", "123456")
            .option("dbtable", "user")
            .load()
        df.show

        // 保存数据
        df.write
            .format("jdbc")
            .option("url", "jdbc:mysql://hadoop102:3306/atcper")
            .option("driver", "com.mysql.cj.jdbc.Driver")
            .option("user", "root")
            .option("password", "123456")
            .option("dbtable", "user1")
            .mode(SaveMode.Append)
            .save()


        val rdd: RDD[User2] = spark.sparkContext.makeRDD(List(User2("zhb", 20), User2("rzl", 23)))
        val ds: Dataset[User2] = rdd.toDS()
        ds.write
            .format("jdbc")
            .option("url", "jdbc:mysql://hadoop102:3306/atcper")
            .option("driver", "com.mysql.cj.jdbc.Driver")
            .option("user", "root")
            .option("password", "123456")
            .option("dbtable", "user2")
            .mode(SaveMode.Append)
            .save()

        //方式 2:通过 jdbc 方法
/*        val props: Properties = new Properties()
        props.setProperty("user", "root")
        props.setProperty("password", "123123")
        ds.write.mode(SaveMode.Append).jdbc("jdbc:mysql://linux1:3306/spark-sql", "user", props)*/

        // TODO 关闭环境
        spark.close()
    }

    case class User2(name: String, age: Int)
}
